package mapreduce;
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.Map;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class MapJoin {

    public static class MyMapper extends Mapper < Object, Text, Text, Text > {
        private Map < String,
        String > dict = new HashMap < > ();

        @Override
        protected void setup(Context context) throws IOException,
        InterruptedException {
            String fileName = context.getLocalCacheFiles()[0].getName();
            //System.out.println(fileName);  
            BufferedReader reader = new BufferedReader(new FileReader(fileName));
            String codeandname = null;
            while (null != (codeandname = reader.readLine())) {
                String str[] = codeandname.split(" ");
                dict.put(str[0], str[2] + " " + str[3]);
            }
            reader.close();
        }
        @Override
        protected void map(Object key, Text value, Context context)
        throws IOException,
        InterruptedException {
            String[] kv = value.toString().split(" ");
            if (dict.containsKey(kv[1])) {
                context.write(new Text(kv[1]), new Text(dict.get(kv[1]) + " " + kv[2]));
            }
        }
    }
    public static class MyReducer extends Reducer < Text, Text, Text, Text > {
        @Override
        protected void reduce(Text key, Iterable < Text > values, Context context)
        throws IOException,
        InterruptedException {
            for (Text text: values) {
                context.write(key, text);
            }
        }
    }

    public static void main(String[] args) throws ClassNotFoundException, IOException, InterruptedException, URISyntaxException {
        Job job = Job.getInstance();
        job.setJobName("mapjoin");
        job.setJarByClass(MapJoin.class);

        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        Path in = new Path("hdfs://localhost:9000/mymapreduce5/in/order_items1");
        Path out = new Path("hdfs://localhost:9000/mymapreduce5/out");
        FileInputFormat.addInputPath(job, in );
        FileOutputFormat.setOutputPath(job, out);

        URI uri = new URI("hdfs://localhost:9000/mymapreduce5/in/orders1");
        job.addCacheFile(uri);

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}